package org.momo.rpc.loadbalance.impl;

import org.momo.rpc.loadbalance.LoadBalancer;
import org.momo.rpc.protocol.RpcRequest;
import org.momo.rpc.provider.model.ProviderRegistryInfo;

import java.util.List;
import java.util.SortedMap;
import java.util.TreeMap;

public class ConsistentHashLoadBalancer implements LoadBalancer {

    private static final int VIRTUAL_NODE_SIZE = 5;

    private int identityHashCode = 0;

    private final TreeMap<Long, ProviderRegistryInfo> virtualInvokers = new TreeMap<>();

    @Override
    public ProviderRegistryInfo loadBalance(RpcRequest request, List<ProviderRegistryInfo> possibleServiceList) {
        int identityHashCode = System.identityHashCode(possibleServiceList);
        if(identityHashCode != this.identityHashCode) {
            virtualInvokers.clear();
            initHashCircle(possibleServiceList);
            this.identityHashCode = identityHashCode;
        }

        long key = hash(request.toString());
        SortedMap<Long, ProviderRegistryInfo> serviceMetaInfos = virtualInvokers.tailMap(key);
        if(serviceMetaInfos.isEmpty()) {
            return virtualInvokers.get(virtualInvokers.firstKey());
        }
        return virtualInvokers.get(serviceMetaInfos.firstKey());
    }

    private void initHashCircle(List<ProviderRegistryInfo> possibleServiceList) {
        for(ProviderRegistryInfo providerRegistryInfo : possibleServiceList) {
            for(int i = 0; i < VIRTUAL_NODE_SIZE; i++) {
                long key = hash(providerRegistryInfo.getServiceName() + i);
                virtualInvokers.put(key, providerRegistryInfo);
            }
        }
    }

    private static long hash(String str) {
        final int p = 16777619;
        long hash = 2166136261L;

        for (int i = 0; i < str.length(); i++)
            hash = (hash ^ str.charAt(i)) * p;

        hash += hash << 13;
        hash ^= hash >> 7;
        hash += hash << 3;
        hash ^= hash >> 17;
        hash += hash << 5;

        if (hash < 0)
            hash = Math.abs(hash);

        return hash;
    }

}
